package com.lobov.run.hadoop.test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.jobcontrol.JobControl;
import org.apache.hadoop.mapred.lib.ChainMapper;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.lobov.containers.MultiPairYale;
import com.lobov.entities.ContainerOfSnipsYale;
import com.lobov.entities.Timer;
import com.lobov.entities.yale.MatrixYale;
import com.lobov.entities.yale.SnipYale;
import com.lobov.utils.Generator;
import com.lobov.utils.Transformer;
import com.lobov.utils.yale.CutterYale;

public class MyJob extends Configured implements Tool {

	public static class Map extends Mapper<LongWritable, Text, Text, Text> {
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			MatrixYale matrix1 = Generator
					.generateUnCompressedHorisontalMatrixYale(2000, 2000, 0.2);
			MatrixYale matrix2 = Generator
					.generateUnCompressedHorisontalMatrixYale(2000, 2000, 0.2);
			ContainerOfSnipsYale container = null;
			try {
				container = CutterYale.cutMatrixHorizontal(matrix2, 3);
			} catch (Exception e) {
				e.printStackTrace();
			}
			MultiPairYale pair1 = new MultiPairYale(matrix1,
					container.getSnipByIndex(0));
			MultiPairYale pair2 = new MultiPairYale(matrix1,
					container.getSnipByIndex(1));
			MultiPairYale pair3 = new MultiPairYale(matrix1,
					container.getSnipByIndex(2));

			SnipYale snip1 = null;
			SnipYale snip2 = null;
			SnipYale snip3 = null;

			try {
				snip1 = (SnipYale) pair1.compute();
				snip2 = (SnipYale) pair2.compute();
				snip3 = (SnipYale) pair3.compute();

			} catch (Exception e) {
				e.printStackTrace();
			}

			context.write(new Text("Merge"), snip1.toText());
			context.write(new Text("Merge"), snip2.toText());
			context.write(new Text("Merge"), snip3.toText());
		}
	}

	public static class Reduce extends Reducer<Text, Text, Text, Text> {
		public void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			System.out.println("MERGING!!!");
			int snipsXLength = 0;
			MultiPairYale pair = null;
			List<SnipYale> snipsList = new ArrayList<>(3);
			
			for (Text val : values) {
				SnipYale snip = Transformer.snipYaleFromText(val);
				snipsList.add(snip.getOrderNumInContainer(), snip);
				snipsXLength = snip.getXlength();
//				System.out.println(snip);
				// context.write(null, new Text(m.toString()));
			}

			ContainerOfSnipsYale cont = new ContainerOfSnipsYale(3,
					snipsXLength);
			cont.setSnipByIndex(snipsList.get(0), 0);
			cont.setSnipByIndex(snipsList.get(1), 1);
			cont.setSnipByIndex(snipsList.get(2), 2);
			
			
			try {
				MatrixYale matrix = cont.toMatrix();
//				System.out.println(matrix);
			} catch (Exception e) {
				e.printStackTrace();
			}
			

		}
	}

	@Override
	public int run(String[] args) throws Exception {
		
		Configuration conf = getConf();
		conf.set("mapreduce.map.max.attempts", "1");
		Job job = new Job(conf, "MyJob");
		job.setJarByClass(MyJob.class);
		Path in = new Path(args[0]);
		Path out = new Path(args[1]);
		FileInputFormat.setInputPaths(job, in);
		FileOutputFormat.setOutputPath(job, out);
		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		job.setNumReduceTasks(1);
		// job.setOutputFormatClass(NullOutputFormat.class); //отменяет вывод в
		// файл
		
		Timer timer = new Timer();
		timer.startTimer();
		int exit = job.waitForCompletion(true) ? 0 : 1;
//		System.exit(job.waitForCompletion(true) ? 0 : 1);
		System.out.println("END !!!!!!!!!!!");
		System.out.println(timer.getTime()/1000 + " сукунд **********");
		System.exit(exit);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new MyJob(), args);

	}

}
